package com.aotain.nyx.synflood;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.scanattack.ScanAttackFlink;
import com.aotain.nyx.statis.AbnStatisTuple;

public class SynfloodForgedImport implements FlatMapFunction<Tuple2<String,SynfloodTuple>, AbnStatisTuple> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = -4037046728657709277L;

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Long> upStreampacket = new HashMap<String,Long>();

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();

	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;

	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();

	@Override
	public void flatMap(Tuple2<String, SynfloodTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		// TODO Auto-generated method stub
		if(value.f0 == null) {
			return;
		}

		upStreamoctets.put(value.f0, value.f1.getUpStreamOctets());
		upStreampacket.put(value.f0, value.f1.getUpStreamPacket());

		try {
			//		InsertIntoData();
			//伪造源攻击判断
			for(Entry<String, Long> entry : upStreampacket.entrySet()) {

				String entryKey = entry.getKey();
				long currenttime = System.currentTimeMillis();

				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];

				if(upStreampacket.get(entryKey) > 600000) {//如果每秒包数大于10000则判断有攻击
					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,out);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,currenttime,out);
					}
				}else {
					//如果没大于门限 时间超过3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey,false);
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
							mapHighFlow.remove(entryKey);
						}
					}
				}

			} 

			// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束，避免漏处理数据
			ArrayList<String> removeItem = new ArrayList<String>();
			for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
				String entryKey = entry.getKey();
				if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {
					removeItem.add(entryKey);
					if(entryKey.endsWith("isFixed")) {
						EndIntoData(entryKey,true);
					} else {
						EndIntoData(entryKey,false);
					}
					attackStartMap.remove(entryKey);
					mapHighFlow.remove(entryKey);
				}
			}

			for(String key  : removeItem) {
				attackLasttimeMap.remove(key);
			}

			removeItem.clear();
			upStreamoctets.clear();
			upStreampacket.clear();
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("SYNFlood Exception",e);;
		}
	}



	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public void InsertIntoData(Entry<String,Long> entry,long time,Collector<AbnStatisTuple> out) {


		String destAreaName = "";
		String destGis = "";
		String sourceAreaName = "";
		String sourceGis = "";
		String sourceAreaCountry = "";
		String sourceAreaId = "";		
		String sourceProvinceName =  "";; //如果省为空，精确到国家
		String sourceAreaCityId = "";
		String sourceAreaProvinceId = "";
		String dip = "";
		String sip = "";
		String dport = "";
		String gis = "";
		String idc = "";
		String desc = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));

		//KEY
		String entryKey = entry.getKey();

		String[] tuplesp = entryKey.split("\\|",-1);
		dip = tuplesp[0];
		sip = "伪造源";
		dport = tuplesp[1];
		idc = tuplesp[2];				
		destAreaName = tuplesp[3];;
		destGis = tuplesp[4];
		String flowDir = tuplesp[5];
		sourceAreaName = "伪造源";
		//				sourceGis = "伪造源";
		sourceAreaCountry = "伪造源";
		sourceAreaId = "伪造源";	
		sourceProvinceName = "伪造源"; //如果省为空，精确到国家
		sourceAreaCityId = "伪造源";
		sourceAreaProvinceId = "伪造源";
		rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport +  "_5";
		
		double maxValue =  upStreamoctets.get(entryKey)/(1024*60);
		mapHighFlow.put(entryKey, maxValue);		
		desc = String.format("SYNFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%f KB/s]", upStreampacket.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();

		row.add(new DataColumn("ROWKEY",rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP",sip));
		row.add(new DataColumn("DESTPORT",dport));
		row.add(new DataColumn("ACCESSTIME",Long.parseLong(accesstime)));
		row.add(new DataColumn("ABRNORMAL", 5));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", upStreampacket.get(entryKey).toString()));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("UPSTREAMOCTETS", upStreamoctets.get(entryKey)));
		row.add(new DataColumn("UPSTREAMPACKET", upStreampacket.get(entryKey)));
		row.add(new DataColumn("ATTTYPE","1"));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);

		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("SYNFLOOD");
		record.setUpStreamOctets(upStreamoctets.get(entryKey));
		record.setUpStreamPacket(upStreampacket.get(entryKey));
		record.setAttackNum(1l);
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(sourceProvinceName);
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(accesstime);
		out.collect(record);

		Logger.getLogger(ScanAttackFlink.class).info(
				String.format("###SynFlood V2 ATTACK START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));


	}

	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void UpdateData(Entry<String,Long> entry,Collector<AbnStatisTuple> out) {


		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";
		String rowKeyAbnormal = "";

		String entryKey = entry.getKey();
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		String[] tuplesp = entryKey.split("\\|",-1);
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuplesp[4];

		dip = tuplesp[0];
		dport = tuplesp[1];
		sip = "伪造源";
		rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_5";
		
		double maxValue = (double)upStreamoctets.get(entryKey)/(double)60/(double)1024 ;

		if(mapHighFlow.containsKey(entryKey))
		{
			double maxValueMem = mapHighFlow.get(entryKey);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(entryKey, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(entryKey, maxValue);
		}
		desc = String.format("SYNFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%f KB/s]", upStreampacket.get(entryKey), maxValue);

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();

		row.add(new DataColumn("UPSTREAMOCTETS", upStreamoctets.get(entryKey)));
		row.add(new DataColumn("UPSTREAMPACKET", upStreampacket.get(entryKey)));
		row.add(new DataColumn("ATTNUM", upStreampacket.get(entryKey).toString()));
		row.add(new DataColumn("DESC", desc));

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);

		Logger.getLogger(ScanAttackFlink.class).info(
				String.format("###SynFlood V2 ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));

		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("SYNFLOOD");
		record.setUpStreamOctets(upStreamoctets.get(entryKey));
		record.setUpStreamPacket(upStreampacket.get(entryKey));
		record.setAttackNum(0l);
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(sourceProvinceName);
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(accesstime);
		out.collect(record);

	}

	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param entryKey
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void EndIntoData(String entryKey,boolean isFixed) {

		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);


		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[1];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_5";	
			desc = String.format("SYNFLOOD固定源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000); 
		} else {
			dip = tuplesp[0];
			dport = tuplesp[1];
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_5";
			desc = String.format("SYNFLOOD伪造源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000);
		}

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();

		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("ENDTIME", Long.parseLong(new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey)+60*1000l)))));

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);

		Logger.getLogger(ScanAttackFlink.class).info(
				String.format("###SynFlood V2 ATTACK END ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));



	}


}
